package com.heima.kafka.listener;

import com.heima.kafka.stream.KafkaStreamListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;

/**
 * 单词计数流式计算程序
 */
@Slf4j
@Component
public class WordCountKafaStreamListener implements KafkaStreamListener<KStream<String,String>> {


    @Override
    public String listenerTopic() {
        return "testInTopic";//入口Topic名称
    }

    @Override
    public String sendTopic() {
        return "testOutTopic"; //出口Topic名称
    }

    /**
     * 流式计算：在单位时间窗口内快速统计出每个单词出现的次数
     */
    @Override
    public KStream<String, String> getService(KStream<String, String> kStream) {

        KStream<String, String> map = kStream.flatMapValues(value->{
                        log.info("[第1阶段]value:{}", value);
                        String[] wordArr = value.split(" ");
                        List<String> wordList = Arrays.asList(wordArr);

                        //第1阶段的目的：获取原始数据，将原始数据转为列表数据，将列表数据流转到第2阶段处理 - 本案例中列表中数据就是每个单词
                        return wordList;
                }).map((key,value)->{
                        //key是空值，value是第1阶段流转过来的列表中每个元素，在本案例中指的就是每个单词
                        log.info("[第2阶段]value:{}", value);
                        //第2阶段的目的：对流转过来的列表数据进行遍历，将每次遍历得到的元素作为将来第3阶段聚合的key
                        return new KeyValue<>(value, null);
                }).groupByKey().windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                .count(Materialized.as("word-count-" + UUID.randomUUID().toString()))
                .toStream()
                .map((key,value)->{
                        //key就是键的对象数据里面包含了键（本案例中键就是单词）, value就是聚合结果数量（本案例中value就是次数）
                        log.info("[第3阶段]key:{},value:{}", key.key(), value);

                        //第3阶段面对：决定指定对应的key和value给SinkProcessor
                        return new KeyValue<>(key.key(), value.toString());
                });

        return map;
    }
}
